package com.hanrt.mr.combineinputformat;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

/**
 * 1.一旦启动这个线程，运行Job
 * 2.本地模式主要用于测试程序是否正确！
 * 3. 报错：
 * 	ExitCodeException exitCode=1: /bin/bash: line 0: fg: no job control
 *
 *作用： 改变了传统的切片方式！将多个小文件，划分到一个切片中！
			适合小文件过多的场景！

RecordReader:  LineRecordReader,一次处理一行，将一行内容的偏移量作为key，一行内容作为value!
			LongWritable	key
			Text value
	切片：   先确定片的最大值maxSize，maxSize通过参数mapreduce.input.fileinputformat.split.maxsize设置！
		流程： a. 以文件为单位，将每个文件划分为若干part
				①判断文件的待切部分的大小 <=  maxSize,整个待切部分作为1part
                ②maxsize < 文件的待切部分的大小 <= 2* maxSize,将整个待切部分均分为2part
				③文件的待切部分的大小 > 2* maxSize,先切去maxSize大小，作为1部分，剩余待切部分继续判断！
		       b. 将之前切分的若干part进行累加，累加后一旦累加的大小超过 maxSize，这些作为1片！
 */
public class WCDriver {
	
	public static void main(String[] args) throws Exception {
		Path inputPath=new Path("e:/mrinput/combine");
		Path outputPath=new Path("e:/mroutput/combine");
	
		//作为整个Job的配置
		Configuration conf = new Configuration();
		// 设置分片的最大值
		conf.set("mapreduce.input.fileinputformat.split.maxsize", "20480000");
		// 设置输入格式 等同于job.setInputFormatClass(ConbineTextInputFormat.class);
		conf.set("mapreduce.job.inputformat.class", "org.apache.hadoop.mapreduce.lib.input.CombineTextInputFormat");

		//保证输出目录不存在
		FileSystem fs=FileSystem.get(conf);
		
		if (fs.exists(outputPath)) {
			fs.delete(outputPath, true);
		}
		
		// ①创建Job
		Job job = Job.getInstance(conf);
		job.setJarByClass(WCDriver.class);
		
		// 为Job创建一个名字
		job.setJobName("wordcount");
		// ②设置Job
		// 设置Job运行的Mapper，Reducer类型，Mapper,Reducer输出的key-value类型
		job.setMapperClass(WCMapper.class);
		job.setReducerClass(WCReducer.class);
		
		// Job需要根据Mapper和Reducer输出的Key-value类型准备序列化器，通过序列化器对输出的key-value进行序列化和反序列化
		// 如果Mapper和Reducer输出的Key-value类型一致，直接设置Job最终的输出类型
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);
		
		// 声明使用ConbineTextInputFormat
		//job.setInputFormatClass(ConbineTextInputFormat.class);
		
		// 设置输入目录和输出目录
		FileInputFormat.setInputPaths(job, inputPath);
		FileOutputFormat.setOutputPath(job, outputPath);
		
		// ③运行Job
		job.waitForCompletion(true);
		
		
	}

}
